package com.atguigu.gmall.realtime.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.app.BaseAppV2;
import com.atguigu.gmall.realtime.bean.TrafficPageViewBean;
import com.atguigu.gmall.realtime.common.Constant;
import com.atguigu.gmall.realtime.util.DateFormatUtil;
import com.atguigu.gmall.realtime.util.FlinkSinkUtil;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.util.Map;

/**
 * @Author lzc
 * @Date 2022/5/2 20:09
 */
public class Dws_02_DwsTrafficVcChArIsNewPageViewWindowApp extends BaseAppV2 {
    public static void main(String[] args) {
        new Dws_02_DwsTrafficVcChArIsNewPageViewWindowApp()
            .init("DwsTrafficVcChArIsNewPageViewWindowApp",
                  4002,
                  1,
                  "DwsTrafficVcChArIsNewPageViewWindowApp",
                  "DwsTrafficVcChArIsNewPageViewWindowApp",
                  Constant.TOPIC_DWD_TRAFFIC_PAGE, Constant.TOPIC_DWD_TRAFFIC_UJ_DETAIL, Constant.TOPIC_DWD_TRAFFIC_UV_DETAIL
            );
    }
    
    @Override
    public void handle(StreamExecutionEnvironment env,
                       Map<String, DataStreamSource<String>> streams) {
        // 1. 多个流合并为一个流
        DataStream<TrafficPageViewBean> pageViewBeanStream = unionOne(streams);
        // 2. 开窗聚合
        SingleOutputStreamOperator<TrafficPageViewBean> aggregatedStream = windowAndAgg(pageViewBeanStream);
        
        // 3. 结果写出到clickhouse中
        writeToClickHouse(aggregatedStream);
    }
    
    private void writeToClickHouse(SingleOutputStreamOperator<TrafficPageViewBean> aggregatedStream) {
        aggregatedStream.addSink(FlinkSinkUtil.getClickHouseSink("gmall2022",
                                                                 "dws_traffic_vc_ch_ar_is_new_page_view_window",
                                                                 TrafficPageViewBean.class
        ));
    }
    
    private SingleOutputStreamOperator<TrafficPageViewBean> windowAndAgg(DataStream<TrafficPageViewBean> stream) {
        return stream
            .assignTimestampsAndWatermarks(
                WatermarkStrategy
                    .<TrafficPageViewBean>forBoundedOutOfOrderness(Duration.ofSeconds(10))
                    .withTimestampAssigner((tpvb, ts) -> tpvb.getTs())
            )
            .keyBy(tpvb -> tpvb.getVc() + "_" + tpvb.getAr() + "_" + tpvb.getCh() + "_" + tpvb.getIs_new())
            .window(TumblingEventTimeWindows.of(Time.seconds(10)))
            .reduce(
                new ReduceFunction<TrafficPageViewBean>() {
                    @Override
                    public TrafficPageViewBean reduce(TrafficPageViewBean v1,
                                                      TrafficPageViewBean v2) throws Exception {
                        v1.setPv_ct(v1.getPv_ct() + v2.getPv_ct());
                        v1.setUj_ct(v1.getUj_ct() + v2.getUv_ct());
                        v1.setSv_ct(v1.getSv_ct() + v2.getSv_ct());
                        v1.setUj_ct(v1.getUj_ct() + v2.getUj_ct());
                        v1.setDur_sum(v1.getDur_sum() + v2.getDur_sum());
                        return v1;
                    }
                },
                new ProcessWindowFunction<TrafficPageViewBean, TrafficPageViewBean, String, TimeWindow>() {
                    @Override
                    public void process(String key,
                                        Context ctx,
                                        Iterable<TrafficPageViewBean> elements,
                                        Collector<TrafficPageViewBean> out) throws Exception {
                        
                        String stt = DateFormatUtil.toYmdHms(ctx.window().getStart());
                        String edt = DateFormatUtil.toYmdHms(ctx.window().getEnd());
                        TrafficPageViewBean trafficPageViewBean = elements.iterator().next();
                        trafficPageViewBean.setStt(stt);
                        trafficPageViewBean.setEdt(edt);
                        trafficPageViewBean.setTs(System.currentTimeMillis());
                        out.collect(trafficPageViewBean);
                        
                    }
                }
            );
        
        
    }
    
    private DataStream<TrafficPageViewBean> unionOne(Map<String, DataStreamSource<String>> streams) {
        // 1. 用于统计会话页面浏览数页面访问时长
        SingleOutputStreamOperator<TrafficPageViewBean> mainStream = streams
            .get(Constant.TOPIC_DWD_TRAFFIC_PAGE)
            .map(json -> {
                JSONObject obj = JSON.parseObject(json);
                JSONObject common = obj.getJSONObject("common");
                JSONObject page = obj.getJSONObject("page");
                String vc = common.getString("vc");
                String ch = common.getString("ch");
                String ar = common.getString("ar");
                String isNew = common.getString("is_new");
                
                Long uvCt = 0L;
                Long svCt = page.getString("last_page_id") == null ? 1L : 0L;
                Long pvCt = 1L;
                Long durSum = page.getLong("during_time");
                Long ujCt = 0L;
                
                
                Long ts = obj.getLong("ts");
                
                
                return new TrafficPageViewBean("", "", vc, ch, ar, isNew,
                                               uvCt, svCt, pvCt, durSum, ujCt,
                                               ts
                );
            });
        // 2. uv
        SingleOutputStreamOperator<TrafficPageViewBean> uvStream = streams
            .get(Constant.TOPIC_DWD_TRAFFIC_UV_DETAIL)
            .map(json -> {
                JSONObject obj = JSON.parseObject(json);
                JSONObject common = obj.getJSONObject("common");
                JSONObject page = obj.getJSONObject("page");
                String vc = common.getString("vc");
                String ch = common.getString("ch");
                String ar = common.getString("ar");
                String isNew = common.getString("is_new");
                
                Long uvCt = 1L;
                Long svCt = 0L;
                Long pvCt = 0L;
                Long durSum = 0L;
                Long ujCt = 0L;
                
                
                Long ts = obj.getLong("ts");
                
                
                return new TrafficPageViewBean("", "", vc, ch, ar, isNew,
                                               uvCt, svCt, pvCt, durSum, ujCt,
                                               ts
                );
            });
        
        // 3. uj
        SingleOutputStreamOperator<TrafficPageViewBean> ujStream = streams
            .get(Constant.TOPIC_DWD_TRAFFIC_UJ_DETAIL)
            .map(json -> {
                JSONObject obj = JSON.parseObject(json);
                JSONObject common = obj.getJSONObject("common");
                JSONObject page = obj.getJSONObject("page");
                String vc = common.getString("vc");
                String ch = common.getString("ch");
                String ar = common.getString("ar");
                String isNew = common.getString("is_new");
                
                Long uvCt = 0L;
                Long svCt = 0L;
                Long pvCt = 0L;
                Long durSum = 0L;
                Long ujCt = 1L;
                
                
                Long ts = obj.getLong("ts");
                
                
                return new TrafficPageViewBean("", "", vc, ch, ar, isNew,
                                               uvCt, svCt, pvCt, durSum, ujCt,
                                               ts
                );
            });
        return mainStream.union(uvStream, ujStream);
    }
}
